package com.visec.hdy.hadoop.storage;

import com.visec.hdy.hadoop.standard.Analysis.Item;

import java.io.IOException;
import java.io.OutputStream;

import org.apache.log4j.Logger;

import com.visec.hdy.hadoop.standard.FSB;
import com.visec.hdy.hadoop.standard.Storage;
import com.visec.hdy.hadoop.standard.WriteContext;

public class DefaultStorageImpl<K, V> implements Storage<K, V> {

	protected OutputStream outputStream;
	protected FSB fsb;
	protected WriteContext<K> context;
	private final Logger log = Logger.getLogger(DefaultStorageImpl.class);

	public DefaultStorageImpl(FSB fsb) {
		this.fsb = fsb;
	}

	@Override
	public void store(Item<K, V> item) {
		if (this.outputStream == null || (!context.write())) {
			this.outputStream = this.fsb.create(context.newFile(item.getKey()));
		}
		try {
			this.outputStream.write(item.getObject().toString().getBytes());
		} catch (IOException e) {
			log.error("", e);
		}
	}

	@Override
	public void setContext(WriteContext<K> context) {
		this.context = context;
	}

	@Override
	public void open() {
		throw new UnsupportedOperationException();
	}

	@Override
	public void flush() {
		try {
			outputStream.flush();
		} catch (IOException e) {
			log.error("", e);
		}
	}

	@Override
	public void close() {
		try {
			if (this.outputStream != null)
				this.outputStream.close();
		} catch (IOException e) {
			log.error("", e);
		}
	}

}
